package com.tjbklx1.rocketMQTest.demo01;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import com.tjbklx1.rocketMQTest.constant.Host;

import java.util.List;
import java.util.UUID;

public class PushConsumerTest {
	private static int count = 0;

	public static void main(String[] args) {
		System.out.print("Push Consumer main start!\n");
		count = 0;
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(Host.PUSH_CONSUMER_GROUP);
		consumer.setInstanceName(UUID.randomUUID().toString());
		consumer.setMessageModel(MessageModel.BROADCASTING);
		consumer.setConsumeMessageBatchMaxSize(32);
		consumer.setNamesrvAddr(Host.HOST);
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		consumer.registerMessageListener(new MessageListenerConcurrently() {
			@Override
			public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
					ConsumeConcurrentlyContext consumeConcurrentlyContext) {
				System.out.print("list count=" + list.size() + "\n");
				for (MessageExt me : list) {
					count++;
					System.out.print("count=" + count + ", msg=" + new String(me.getBody()) + "\n");
				}
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
			}
		});
		try {
			consumer.subscribe(Host.TOPIC, "*");
			consumer.start();
			System.out.print("Push Consumer Started!\n");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}
